package org.example.finance.monitor;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.example.finance.bean.TransactionRecord;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.Date;

/**
 * @author huangqihan
 * @date 2021/2/22
 */
public class CurrentDayMonitor {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<TransactionRecord> inputStream = env.addSource(new DataSource());
        inputStream.print();

        Long timeOffset = getTimeOffset();

        // 1.当天交易总笔数
        SingleOutputStreamOperator<Long> result0 = inputStream.keyBy("type")
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.milliseconds(timeOffset)))
                .aggregate(new AggregationFunc());
//        result0.print();

        // 2.当天交易近五分钟峰值数
        SingleOutputStreamOperator<Long> result1 = inputStream.keyBy("type")
                .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.seconds(1)))
                .aggregate(new AggregationFunc());

        result1.print();
        env.execute();
    }

    static class DataSource implements SourceFunction<TransactionRecord> {
        private boolean running = true;

        @Override
        public void run(SourceContext<TransactionRecord> sourceContext) throws Exception {
            long key = 0;
            while (running) {
                sourceContext.collect(new TransactionRecord(++key, System.currentTimeMillis(), TransactionRecord.Type.DEAL));
                Thread.sleep((long) (Math.random() * 1000 * 2));
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

    static class AggregationFunc implements AggregateFunction<TransactionRecord, Long, Long> {

        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(TransactionRecord transactionRecord, Long acc) {
            long sum = acc + 1;
//            System.out.println("Event Time: " + new Date() + ", Sum: " + sum);
            return sum;
        }

        @Override
        public Long getResult(Long acc) {
            return acc;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    public static Long getEndOfDay(Date date) {
        LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(date.getTime()), ZoneId.systemDefault());
        LocalDateTime endOfDay = localDateTime.with(LocalTime.MAX);
        Date max = Date.from(endOfDay.atZone(ZoneId.systemDefault()).toInstant());
        return max.getTime();
    }

    public static Long getStartOfDay(Date date) {
        LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(date.getTime()), ZoneId.systemDefault());
        LocalDateTime startOfDay = localDateTime.with(LocalTime.MIN);
        Date min = Date.from(startOfDay.atZone(ZoneId.systemDefault()).toInstant());
        return min.getTime();
    }

    public static Long getTimeOffset() {
        long now = System.currentTimeMillis();
        Long start = getStartOfDay(new Date());
        return start - now;
    }
}
